package com.cotte.flow.hbase.dao;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by zhuang on 2018/12/28.
 */
@Component
public class GraphTableDaoImpl {

    private String tableName = "step:brushcard_graphtable";
    private Table table = null;
    private Map<String,String> row = null;

    @Autowired
    private Connection connection;

    public ArrayList getData(String small,String big){
        //String key = small.substring(0,odid.length()-7);
        Scan s = new Scan();
        s.setStartRow(small.getBytes());
        s.setStopRow(big.getBytes());
        //s.setFilter(new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator(key+".*")));
        //
        ArrayList result = new ArrayList();
        ResultScanner scanner = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            scanner = table.getScanner(s);
            for(Result one:scanner){
                model(one);
                if(row.get("workhours").equals("0")){
                    continue;
                }
                row.put("time",formatDate(row.get("key").substring(row.get("key").length()-6)));
                result.add(row);
            }
            scanner.close();
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

    public ArrayList filterData(String filter){
        Scan s = new Scan();
        s.setFilter(new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator(filter)));
        //
        ArrayList result = new ArrayList();
        ResultScanner scanner = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            scanner = table.getScanner(s);
            for(Result one:scanner){
                result.add(model(one));
            }
            scanner.close();
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }

    public void updateRow(String odid,String workhours,String records,String range) {
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(odid.getBytes("utf-8"));
            put.addColumn("info".getBytes(),"workhours".getBytes(),workhours.getBytes());
            put.addColumn("info".getBytes(), "records".getBytes(), records.getBytes());
            if(!isEmpty(range)){
                put.addColumn("info".getBytes(), "range".getBytes(), range.getBytes());
            }
            table.put(put);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /*
     *  获取工时范围
     */
    public String hoursRange(String odid){
        String res = null;
        try {
            table = connection.getTable(TableName.valueOf("step:brushcard_ref"));
            Get get = new Get(odid.getBytes("utf-8"));
            Result result = table.get(get);
            if(!result.isEmpty()){
                res = Bytes.toString(result.getValue("range".getBytes(),"range".getBytes()));
            }
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return res;
    }

    /*
    * 计算一天平均工时
    * */
    public void addDayHours(String odid,String records) {
        com.alibaba.fastjson.JSONArray hours = JSON.parseArray(records);
        String key = odid.substring(0,odid.length()-7);
        System.out.print("start : (计算工时) " + System.currentTimeMillis() +" -- ");
        String range = hoursRange(key);
        JSONArray hoursRange = JSON.parseArray(range);
        System.out.print("1 : "+ key +"  " + System.currentTimeMillis()+ " -- ");
        //
        Integer count = 0;
        Integer sum = 0;
        for(int i = 0;i<hours.size();i++) {
            Integer val = hours.getIntValue(i);
            if(null != hoursRange){
                Integer min = hoursRange.getInteger(0);
                Integer max = hoursRange.getInteger(1);
                if(min>val || val>max){// 范围之外
                    continue;
                }
                sum += val;
                count++;
            }else{// 未被统计
                sum += val;
                count++;
            }
        }
        String workhours = count>0?String.valueOf(Math.round(sum/count)):"0";
        updateRow(odid ,workhours ,records,range);
        System.out.print("end : (计算工时)  " + System.currentTimeMillis() +"\n");
    }

    /*
    * 如果workhours为空,则根据records统计平均工时
    * @param isall : true (更新所有)
    * */
    public void addAllHours(String isall) {
        Scan s = new Scan();
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            ResultScanner scanner = table.getScanner(s);
            for(Result one:scanner){
                model(one);
                if(!isEmpty(row.get("records"))){
                    if(!isEmpty(isall) || isEmpty(row.get("workhours"))){
                        addDayHours(row.get("key"),row.get("records"));
                    }
                }
            }
            System.out.print("complete : (计算工时)  \n");
            scanner.close();
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /*
    * model-row
    * */
    private Map<String, String> model(Result one) {
        row = new HashMap<String,String>();
        row.put("key",Bytes.toString(one.getRow()));
        row.put("workhours", Bytes.toString(one.getValue("info".getBytes(),"workhours".getBytes())));
        row.put("records",Bytes.toString(one.getValue("info".getBytes(),"records".getBytes())));
        row.put("range",Bytes.toString(one.getValue("info".getBytes(),"range".getBytes())));
        return row;
    }

    private String formatDate(String date){
        DateFormat dateFormat = new SimpleDateFormat("yyMMdd");
        String res = "";
        try {
            res = String.valueOf(dateFormat.parse(date).getTime());
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return res;
    }

    private Boolean isEmpty(String val){
        if(null == val || val.isEmpty()) {
            return true;
        }else{
            return false;
        }
    }
}

